package com.changgou.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.changgou.canal.config.RabbitMQConfig;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;


/**
 * Created by 柱子 on 2020/10/18 21:14
 */
@CanalEventListener
//声明当前类为Canal监听类
public class BusinessListener {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * @param entryType 当前操作数据库的类型
     * @param rowData   当前操作数据库的数据
     */
    //监听的数据库和数据表
    @ListenPoint(schema = "changgou_business", table = "tb_ad")
    public void adUpdate(CanalEntry.EntryType entryType, CanalEntry.RowData rowData) {
        //监听进入该方法
        System.out.println("数据库广告表发生改变");
        //获取改变之前的数据
        List<CanalEntry.Column> rowDataList = rowData.getAfterColumnsList();

        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
        for (CanalEntry.Column c : beforeColumnsList) {
          //  System.out.println("改变前的 数据:" + c.getName() + "::" + c.getValue());
        }

        //rowData.getBeforeColumnsList().forEach(c -> System.out.println("改变前的 数据:" + c.getName() + "::" + c.getValue()));

        //获取改变之后的数据
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
      //  for (CanalEntry.Column column : afterColumnsList) {

           // System.out.println("改变之后的数据:"+column.getName()+"::"+column.getValue());
      //  }


        //发送消息到RabbitMQ

        for (CanalEntry.Column column : afterColumnsList) {

            if ("position".equals(column.getName())){
                System.out.println("发送数据到RabbitMQ   "+column.getValue());

                //发送
                rabbitTemplate.convertAndSend("", RabbitMQConfig.AD_UPDATE_QUEUE,column.getValue());

            }
        }

    }

}
